package com.haizhi.connectedcomponents

import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo1 {
  case class Person(name: String, age: Int)
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[1]").setAppName("CC")

    val sc: SparkContext = new SparkContext(sparkConf)

    val edgelines = sc.textFile("graphdata/links.csv")

    val edges:RDD[Edge[String]] = edgelines.map({ line =>
      val row = line.split(",")
      Edge(row(0).toInt, row(1).toInt, row(2))
    })

    val peoplelines: RDD[String] = sc.textFile("graphdata/people.csv")

    val peopleRDD = peoplelines.map(line => line.split(","))
      .map(row => (row(0).toLong, Person(row(1), row(2).toInt)))

    val graph: Graph[Person, String] = Graph(peopleRDD, edges)

    val cc: Graph[VertexId, String] = graph.stronglyConnectedComponents(100)

    println("triplets")
//    ((1,1),(2,2),friend) => ((srcId,srcAttr),(dstId,dstAttr),attr)
    cc.triplets.foreach(println)

//    println("triplets.srcId")
//    cc.triplets.map(x=>x.srcId).foreach(println)
//
//    println("triplets.dstId")
//    cc.triplets.map(x=>x.dstId).foreach(println)

//    println("triplets.srcAttr")
//    cc.triplets.map(x=>x.srcAttr).foreach(println)
//
//    println("triplets.dstAttr")
//    cc.triplets.map(x=>x.dstAttr).foreach(println)

        println("triplets.attr")
        cc.triplets.map(x=>x.attr).foreach(println)



  }

}
